package com.kingsoft.dc.khaos.module.spark.sink

import java.util
import java.util.Properties

import com.kingsoft.dc.khaos.KhaosContext
import com.kingsoft.dc.khaos.extender.meta.model.col.DmTableColumn
import com.kingsoft.dc.khaos.innertype.Schema
import com.kingsoft.dc.khaos.module.spark.constants.CommonConstants
import com.kingsoft.dc.khaos.module.spark.metadata.sink.MppSinkConfig
import com.kingsoft.dc.khaos.module.spark.request.model.JdbcConnectEntity
import com.kingsoft.dc.khaos.module.spark.util.{DataframeUtils, MetaUtils}
import com.kingsoft.dc.khaos.util.Logging
import org.apache.spark.sql.DataFrame
import org.json4s.jackson.JsonMethods.{compact, render}
import org.json4s.{DefaultFormats, JsonAST}

/**
  * create by yansu on 2019/08/09 15:38
  */
class MppDirectSink extends SinkStrategy with Logging {
  private var _kc: KhaosContext = null

  private var connectEntity: JdbcConnectEntity = null
  private var _mpp_Config: MppSinkConfig = null
  private var _mppColsInfo: util.List[DmTableColumn] = null

  // 物理地址
  private var _host: String = null
  private var _port: String = null
  private var _username: String = null
  private var _password: String = null
  private var _instansename: String = null

  /** 数据输出 */
  override def sink(kc: KhaosContext,
                    module_id: String,
                    config: JsonAST.JObject,
                    schema: Schema,
                    dataSet: DataFrame): Any = {

    // 解析config
    implicit val formats = DefaultFormats
    val mppSinkConfig = config.extract[MppSinkConfig]
    _mpp_Config = mppSinkConfig
    _kc = kc

    //权限校验
    MetaUtils.checkWriteAuth(kc,
      mppSinkConfig.db_name,
      mppSinkConfig.table_name,
      mppSinkConfig.extender.auth.clazz,
      compact(render(mppSinkConfig.extender.auth.params)))

    initMetaData()
    initJdbcConnectInfo()

    // 重新构建df，重命名并且添加默认值
    val decimal_switch: Boolean = _kc.conf.getBoolean(CommonConstants.GREENPLUM_DECIMAL_SWITCH, false)
      log.info(s"mppDirect decimal_switch: $decimal_switch")
      val targetData = DataframeUtils.buildNewDataframeGreenplum(null, null, _mpp_Config.db_name, _mppColsInfo, _mpp_Config.extract_fields, dataSet, connectEntity,decimal_switch)._1
      if (targetData.first() != null) {
      log.info("DataFrame转换成功...")
    }

    // 通过jdbc直连方式获取dataframe
    val driver = "org.postgresql.Driver"
    val prop = new Properties
    prop.put("driver", driver)
    prop.put("user", _username)
    prop.put("password", _password)
    log.info("driver=>" + driver)

    var savemode = mppSinkConfig.write_option.trim.toLowerCase
    if (savemode == "" || savemode == null) {
      // 如果前端没传,则默认使用Overwrite
      savemode = "Overwrite"
    }
    // 模式名.表名
    val schemaAndTableName = _mpp_Config.db_name + "." + getMppTableName()
    log.info("开始写入MPP数据表...")
    targetData.write.mode(savemode).jdbc(getConnectUrl(),
      schemaAndTableName,
      prop)
  }

  /**
    * 初始化写入的数据表的元数据信息
    */
  def initMetaData() = {

    log.info("初始化物理地址...")
    val dbname = _mpp_Config.db_name
    val tblname = _mpp_Config.table_name
    val className = _mpp_Config.extender.meta.clazz

    val mppConnect = MetaUtils.getMPPMeta(_kc,
      dbname,
      tblname,
      className,
      compact(render(_mpp_Config.extender.meta.params)),
      this).getDsMppConnect
    _mppColsInfo = MetaUtils.getMPPMeta(_kc, dbname, tblname, className, compact(render(_mpp_Config.extender.meta.params)), this).getColumnEntiy
    _host = mppConnect.getHost
    _port = mppConnect.getPort
    _username = mppConnect.getUsername
    _password = mppConnect.getPassword
    _instansename = mppConnect.getInstanceName
  }

  /**
    * 初识化JDBC连接
    */
  def initJdbcConnectInfo() = {
    log.info("初始化JDBC...")
    val url = getConnectUrl()
    val user = _username
    val password = _password
    val schema = _mpp_Config.db_name
    this.connectEntity = new JdbcConnectEntity(url,
      user,
      password,
      schema,
      getMppTableName())
  }

  /**
    * 获取mpp连接url
    *
    * @return
    */
  def getConnectUrl(): String = {
    val host = _host
    val port = _port
    val dbName = _instansename
    val url = s"jdbc:postgresql://${host}:${port}/${dbName}"
    log.info("url ==> " + url)
    url
  }

  /**
    * 获取mpp内部表名
    *
    * @return
    */
  def getMppTableName() = {
    val mppTable = _mpp_Config.table_name
    mppTable.toLowerCase
  }
}
